package com.dynomedia.esearch.util.groupkeycache;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import com.dynomedia.esearch.util.groupkeycache.util.ConsumerThreadPool;
import com.dynomedia.esearch.util.groupkeycache.util.GkcConfig;

/**
 * @author Administrator
 * @version 1.0
 * @created 19-十月-2009 16:50:13
 */
public class GroupkeyConsumer implements Runnable {
	
	private static final Log logger = LogFactory.getLog(GroupkeyConsumer.class);
	
	/**
	 * 线程的休眠时间等同于groupkey的sort时间
	 */
	private static final long sleepingTime = GkcConfig.getInstance().getGroupkeySortInterval()*1000;		

	/**
	 * 静态初始化GroupkeyConsumer
	 */
	static{
		for(int i=0;i<GkcConfig.getInstance().getGroupkeyConsumerThreadCount();i++){
			ConsumerThreadPool.submit(new GroupkeyConsumer());
		}
		if(logger.isInfoEnabled()) logger.info("GroupkeyConsumer init complete,groupkeyConsumerThreadCount = "+GkcConfig.getInstance().getGroupkeyConsumerThreadCount());		
	}

	private GroupkeyConsumer(){}
	
	public static void start(){}

	/**
	 * GroupkeyConsumer的线程调度方式应保证在每一个时间点，一个Groupkey对象只能由一个特定GroupkeyConsumer处理。
	 */
	public void run(){
		while(ConsumerThreadPool.isRunning()){
			try{
				Integer groupkeysQueueSize = GroupkeyCache.getGroupkeysQueue().size();
				for(int i=0;i<groupkeysQueueSize;i++){
					Groupkey groupkey = GroupkeyCache.getGroupkeysQueue().take();
					if(groupkey.needSort()){
						groupkey.sort();
					}else{
						if(logger.isDebugEnabled()) logger.debug("GroupkeyConsumer no need to flush for Groupkey："+groupkey);
					}
					GroupkeyCache.getGroupkeysQueue().put(groupkey);					
				}
				Thread.sleep(sleepingTime);
			}catch(Exception e){
				logger.error("GroupkeyConsumer run Error!",e);
				continue;
			}
		}
	}

}